package com.longyb.mylive.client;



import com.longyb.mylive.server.handlers.ChunkDecoder;
import com.longyb.mylive.server.handlers.ChunkEncoder;
import com.longyb.mylive.server.handlers.HandShakeDecoder;
import com.longyb.mylive.server.handlers.RtmpMessageHandler;
import com.longyb.mylive.server.rtmp.messages.RtmpMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioChannelOption;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import lombok.extern.slf4j.Slf4j;


import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RtmpClient {
    private static final EventLoopGroup GLOBAL_WORK_GROUP = new NioEventLoopGroup(10);
    private static final EventLoopGroup GLOBAL_HAND_GROUP = new NioEventLoopGroup(10);
    private Channel channel;
    private int connRetry = 3;


    /**
     * 建立一个连接，共享Eventloop
     *
     * @param host
     * @param port
     */
    /**
     * 打开连接
     * @param host
     * @param port
     */
    public void connect(String host, int port) {

        // 配置客户端NIO线程组
        //首先创建客户端处理I/O读写的NioEventLoop Group线程组
        try {

            //继续创建客户端辅助启动类Bootstrap，随后需要对其进行配置。
            //与服务端不同的是，它的Channel需要设置为NioSocketChannel
            //然后为其添加handler。
            // 在初始化它的时候将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件。
            Bootstrap bootstrap = new Bootstrap();


            bootstrap.group(GLOBAL_WORK_GROUP);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.option(NioChannelOption.TCP_NODELAY, true);
            bootstrap.option(NioChannelOption.SO_REUSEADDR, true);
            bootstrap.option(NioChannelOption.SO_KEEPALIVE, false);
            bootstrap.option(NioChannelOption.SO_RCVBUF, 2048);
            bootstrap.option(NioChannelOption.SO_SNDBUF, 2048);
            //增加编解码工具
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                    super.channelInactive(ctx);

                }

                @Override
                public void initChannel(SocketChannel ch) {

                    /**
                     * maxFrameLength最大包长
                     * lengthFieldOffset 长度字段偏移，第一个字段不是长度字段
                     * lengthFieldLength 长度字段字节数
                     * lengthAdjustment 长度字段修正，针对有特定头的
                     * initialBytesToStrip 丢弃字节
                     * failFast 快速失败
                     *
                     * 坑，默认lengthAdjustment不算长度字段
                     */
                    ch.pipeline().addLast("logging", new LoggingHandler(LogLevel.INFO));

                    //跳分隔符，快速失败

                        ch.pipeline()
                                .addLast(new HandShakeEncoder())
                                .addLast(new ChunkDecoder())
                                .addLast(new ChunkEncoder())
                                .addLast(GLOBAL_HAND_GROUP,new RtmpClientHandler());
                    ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(10, 10, 10, TimeUnit.MINUTES));

                }
            });
            //然后调用同步方法等待连接成功。
            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();

            /**
             * 监听
             */
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {

                    //任务完成
                    if (channelFuture.isDone()) {
                        if (channelFuture.isSuccess()) {
                            // 成功
                            log.info("客户端连接成功:{},{}", host, port);
                            channel = channelFuture.channel();
                        } else if (channelFuture.isCancelled()) {
                            // 被取消
                            log.info("客户端连接取消:{},{}", host, port);
                        } else {
                            //重连交给后端线程执行
                            if ((connRetry--) > 0) {
                                channelFuture.channel().eventLoop().schedule(() -> {
                                    log.error("重连服务端...");
                                    connect(host, port);
                                }, 3000, TimeUnit.MILLISECONDS);
                            }
                            // 异常
                            log.error("客户端连接异常,3秒后重试:{},{},{}", host, port, connRetry, channelFuture.cause());
                        }
                    }
                }
            });
            channel = channelFuture.channel();
        } catch (InterruptedException e) {
            log.error("连接中断", e);
        }

    }


    public void setCallback() {

    }


    public boolean isConnect() {
        if (null == channel) {
            return false;
        }
        return channel.isActive();
    }

    /**
     * @param buf
     */
    public void write(Object buf) {
        if (null == channel) {
            log.warn("channel is null or close");
        } else {
            channel.writeAndFlush(buf);
        }

    }
    public void close(){
        if(channel !=null && channel.isActive()){
            channel.close();
        }
    }

    /**
     * 握手处理
     */
    public void handshake(){
        //1. 发送c0,c1

        //2. 接收s0,s1,s2->回复 c2



    }

    public static void main(String[] args) throws Exception {
        RtmpClient client=new RtmpClient();
        client.connect("127.0.0.1",1935);
//        client.write();


//        RtmpClient client = RtmpClient.build("8.218.32.7", 8200, new FrameHandler() {
//
//        });
    }
}
//            @Override
//            public void onFrame(ChannelHandlerContext ctx, Frame f) {
//
//                log.info("Server received: {}", JSON.toJSONString(f));
//            }
//        });

//        JTT808Client client = JTT808Client.build("58.250.212.2", 8100, new FrameHandler() {
//            //            @Override
//            public void onFrame(ChannelHandlerContext ctx, Frame f) {
//
//                log.info("Server received: {}", JSON.toJSONString(f));
//            }
//        });
//        HLXTFrame frame=new HLXTFrame();
//        frame.setMfr("3G");
//        frame.setDid("8031230124");
//        frame.setBody("TK,REC");
//        client.write(frame);
//        Thread.sleep(1000);

//        for (int i=0;i<20;i++){
//            JTT808Frame frame=new JTT808Frame();
////            frame.setAudioFrame(false);
////            frame.setMfr("3G");
////            frame.setDid("8010153041");
////            frame.setBody("GS1,1,5,460,0,9773,4182,90");
//
//            client.write(frame);
//            Thread.sleep(1000);
//        }
//
//        for (int i=0;i<2;i++){
//            JTT808Frame frame=new JTT808Frame();
////            frame.setAudioFrame(true);
//////            frame.setMfr("3G");
////            frame.setDid("8010153041");
////            frame.setBody("UD2,091220,103259,V,00.000000,N,000.000000,E,0,0,0,0,100,20,0,0,00000009,1,5,460,0,9773,4113,90,0");
//
//            client.write(frame);
//            Thread.sleep(1000);
//        }

//        HLXTFrame frame=new HLXTFrame();
//        for (int i=0;i<3;i++){
//            frame.setAudioFrame(true);
//            frame.setDid("8010153041");
//            frame.setFileSize(0x00003e66);
//            frame.setSliceCount(0x06);
//            frame.setSliceSeq(0x01);
//            frame.setSliceSize((short)0x02);
//            byte[] data=Hex.decodeHex("4eb8");
//            frame.setSliceData(data);
//            client.write(frame);
//        }



